package com.fire.market.mq;

import com.alibaba.fastjson.JSONObject;
import com.fire.component.catchlog.utils.LoggerUtils;

import com.fire.component.extension.Assert;
import com.fire.market.product.api.mq.ItemStockMqMessage;
import org.slf4j.Logger;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.UUID;

/**
 * @author zwd
 * @date 2021/6/27
 * @email zwd@hhh.com
 */
@Component
public class MQProducer implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback, InitializingBean {
    Logger LOGGER = LoggerUtils.Logger(MQProducer.class);

    @Resource
    private RabbitTemplate rabbitTemplate;

    private static ThreadLocal<Object> threadLocalMsg = new ThreadLocal<>();
    /**
     * 系统配置
     */
    @Value("${mq.producer.retry-count}")
    private Integer mqRetryCount;

    public void sendMessage(String exchangeName, String routingKey, Object message){
        Assert.notNull(exchangeName, "exchangeName is null");
        Assert.notNull(routingKey, "routingKey is null");
        Assert.notNull(message, "message is null");

        CorrelationDataExtra correlationDataExtra = this.correlationData(message);
        correlationDataExtra.setExchange(exchangeName);
        correlationDataExtra.setRoutingKey(routingKey);
        correlationDataExtra.setMessage(message);

        LOGGER.info("发送MQ消息，消息ID：{}，消息体:{}, exchangeName:{}, routingKey:{}",
                correlationDataExtra.getId(), JSONObject.toJSONString(message), exchangeName, routingKey);

        // 发送消息
        this.convertAndSend(exchangeName, routingKey, message, correlationDataExtra);
    }

    /**
     * 用于实现消息发送到RabbitMQ交换器后接收ack回调。
     * 如果消息发送确认失败就进行重试。
     *
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        // 消息回调确认失败处理
        if (!ack && correlationData instanceof CorrelationDataExtra){
            CorrelationDataExtra correlationDataExtra = (CorrelationDataExtra) correlationData;

            //消息发送失败，进行重试，重试过后不能成功就记录到数据库
            if (correlationDataExtra.getRetryCount() < mqRetryCount){
                LOGGER.info("MQ消息发送失败，消息重发，消息ID：{}，重发次数：{}，消息体:{}", correlationDataExtra.getId(),
                        correlationDataExtra.getRetryCount(), JSONObject.toJSONString(correlationDataExtra.getMessage()));
                // 将重试次数加一
                correlationDataExtra.setRetryCount(correlationDataExtra.getRetryCount() + 1);
                // 重发发消息
                this.convertAndSend(correlationDataExtra.getExchange(), correlationDataExtra.getRoutingKey(),
                        correlationDataExtra.getMessage(), correlationDataExtra);
            }else {
                // TODO 保存消息到数据库
                saveToDb(threadLocalMsg.get(),"消息没有正确落到交换器，消息体为:" + JSONObject.toJSONString(correlationDataExtra.getMessage()));
            }
        }else {
            LOGGER.info("消息发送成功,消息ID:{}", correlationData.getId());
        }
    }

    /**
     * 发送消息
     *
     * @param exchange        交换机名称
     * @param routingKey      路由key
     * @param message         消息内容
     * @param correlationData 消息相关数据（消息ID）
     * @throws AmqpException
     */
    private void convertAndSend(String exchange, String routingKey, final Object message, CorrelationData correlationData) throws AmqpException {
        try {
            threadLocalMsg.set(message);
            rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
        } catch (Exception e) {
            LOGGER.error("MQ消息发送异常，消息ID：{}，消息体:{}, exchangeName:{}, routingKey:{}",
                    correlationData.getId(), JSONObject.toJSONString(message), exchange, routingKey, e);

            // TODO 保存消息到数据库
            saveToDb(message,"消息发送异常");
        }
    }

    private void saveToDb(Object message, String extra) {
        if (message instanceof ItemStockMqMessage) {
            ItemStockMqMessage itemStockMqMessage = (ItemStockMqMessage) message;
        }
    }

    /**
     * 消息相关数据（消息ID）
     *
     * @param message
     * @return
     */
    private CorrelationDataExtra correlationData(Object message) {

        return new CorrelationDataExtra(UUID.randomUUID().toString(), message);
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchangeName, String RoutingKey) {
        LOGGER.error("MQ消息发送失败，replyCode:{}, replyText:{}，exchange:{}，routingKey:{}，消息体:{}",
                replyCode,replyText,exchangeName,RoutingKey, message.getBody());
        // TODO 保存消息到数据库
        saveToDb(threadLocalMsg.get(),"消息没有正确落到磁盘,消息体为：" + JSONObject.toJSONString(message.getBody()));
    }
}
